package com.aqielife.demo.controller;


import com.aqielife.common.R;
import com.aqielife.common.thread.NonBlockingThread;
import com.aqielife.demo.entity.City;
import com.aqielife.demo.service.CityService;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.NonBlocking;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author aqie
 * @date 2022/02/04 20:23
 * @desc
 */
@Slf4j
@RestController
@RequestMapping(value = "/city")
public class CityWebFluxController {

  @Autowired
  @Qualifier("cityServiceMongoImpl")
  private CityService cityService;

  @GetMapping(value = "/{id}")
  public Mono<City> findCityById(@PathVariable("id") String id) {
    return cityService.findCityById(id);
  }

  @GetMapping()
  public Flux<City> findAllCity() {
    return cityService.findAllCity();
  }

  @PostMapping()
  public Mono<City> saveCity(@RequestBody City city) {
    return cityService.save(city);
  }

  @PutMapping()
  public Mono<City> modifyCity(@RequestBody City city) {
    return cityService.modifyCity(city);
  }

  @DeleteMapping(value = "/{id}")
  public void deleteCity(@PathVariable("id") String id) {
    cityService.deleteCity(id);
  }

  @GetMapping("webClient")
  public Mono<R> webClient() {
    Mono<R> resp = WebClient.create("http://api.hscxjy.com/")
    .method(HttpMethod.GET)
    .uri("aqie-edu/gradeclass/tenantListTree?tenantId={tenantId}","936956")
    .cookie("token","xxxx")
    .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
    .retrieve().bodyToMono(R.class);
    // log.info("{}", resp.block());
    return resp;
  }

  public static void main(String[] args) throws InterruptedException {
    Thread th = new NonBlockingThread(new Runnable() {
      @Override
      public void run() {
        if (Thread.currentThread() instanceof NonBlocking){
          System.out.println("====");
        }
        CountDownLatch countDownLatch = new CountDownLatch(1);
        AtomicReference reference = new AtomicReference<>();
        Mono.create(
        monoSink -> {
          monoSink.success("sasa");
        }
        ).publishOn(Schedulers.boundedElastic()).subscribe(
        s->{
          reference.set(s);
          countDownLatch.countDown();
        }
        );
        try {
          countDownLatch.await();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        System.out.println(reference.get());
      }
    }, "sasa");
    th.start();

    Thread.sleep(5000L);
  }

  /**
   * Future进行并行异步调用
   * @return
   * @throws InterruptedException
   * @throws ExecutionException
   */
  @GetMapping("demo")
  public R<String> demo() throws InterruptedException, ExecutionException {
    ExecutorService executorService = Executors.newFixedThreadPool(10);

    long startTime = System.currentTimeMillis();

    // 城市基本信息
    FutureTask<String> cityInfoFutureTask = new FutureTask<>(new Callable<String>() {
      @Override
      public String call() throws Exception {
        return cityService.getCityInfo();
      }
    });
    executorService.submit(cityInfoFutureTask);

    Thread.sleep(200); //模拟主线程其它操作耗时

    FutureTask<String> medalInfoFutureTask = new FutureTask<>(new Callable<String>() {
      @Override
      public String call() throws Exception {
        return cityService.getCityMedal();
      }
    });
    executorService.submit(medalInfoFutureTask);

    // 阻塞调用
    String cityInfo = cityInfoFutureTask.get();//获取城市信息结果
    String medalInfo = medalInfoFutureTask.get();//获取勋章信息结果
    log.info("{} {}", cityInfo, medalInfo);
    String str = "总共用时" + (System.currentTimeMillis() - startTime) + "ms";
    return R.data(str);
  }

  @GetMapping("demo2")
  public R<String> demo2() throws InterruptedException, ExecutionException, TimeoutException {
    // 使用自定义线程池
    ExecutorService executor = Executors.newCachedThreadPool();
    long startTime = System.currentTimeMillis();
    // 1. 城市基本信息
    CompletableFuture<String> completableCityInfoFuture = CompletableFuture.supplyAsync(() -> cityService.getCityInfo(), executor);
    CompletableFuture<String> completableCityHisFuture =
    completableCityInfoFuture.thenApplyAsync((city) -> cityService.getCityInfoHistory(city));

    Thread.sleep(200); //模拟主线程其它操作耗时


    CompletableFuture<String> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> cityService.getCityMedal(), executor);

    String cityInfo = completableCityInfoFuture.get(2,TimeUnit.SECONDS);//获取个人信息结果
    String historyInfo = completableCityHisFuture.get();
    String medalInfo = completableMedalInfoFuture.get();//获取勋章信息结果
    log.info("{} {} {}", cityInfo, historyInfo, medalInfo);

    String str = "总共用时" + (System.currentTimeMillis() - startTime) + "ms";
    executor.shutdown(); // 线程池需要关闭
    return R.data(str);
  }
}
